from pyspark.sql import SparkSession
import re

"""
@version:
@software: PyCharm
@file: 
@time: 
"""

spark = SparkSession.builder.master(master="local").appName("test").getOrCreate()

sc = spark.sparkContext


# pyspark读取txt文件并去除符号，处理文本文件的一些技巧


# 1、将文本内容读取出来并将
def removePunctuation(text):
    """Removes punctuation, changes to lower case, and strips leading and trailing spaces.

    Note:
        Only spaces, letters, and numbers should be retained.  Other characters should should be
        eliminated (e.g. it's becomes its).  Leading and trailing spaces should be removed after
        punctuation is removed.

    Args:
        text (str): A string.

    Returns:
        str: The cleaned up string.
    """
    # 替换不是已字母数字的字符替换为空，     替换内容，并将大写装小写
    return re.sub("[^a-zA-Z0-9 ]", "", text.strip(" ").lower())


# 文件的路径
filepath = "E:\\Python\\pyspark_demo01\\pyspark_data\\shakespear.txt"

# 读取文件
shakespeareRDD = sc.textFile(filepath).map(removePunctuation)
# print("查看处理过的文档:{}".format(shakespeareRDD.collect()))

####################################################################

# 切词操作
shakespeareWordsRDD = shakespeareRDD.flatMap(lambda a: a.split(" "))

# 执行分词之后的操作
shakespeareWordCount = shakespeareWordsRDD.count()
print("查看词出现的前五个：{}".format(shakespeareWordsRDD.top(5)))
print("查看总共有多少个词：{}".format(shakespeareWordCount))

####################################################################

# 移除空白值
shakeWordsRDD = shakespeareWordsRDD.filter(lambda word: len(word) > 0)
shakeWordCount = shakeWordsRDD.count()
print("打印移除空白字符之后剩余的字符总数：{}".format(shakeWordCount))


####################################################################

# 用科学技术法将字词数计数并排序
# 一个 RDD函数
def wordCount(wordListRDD):
    """Creates a pair RDD with word counts from an RDD of words.

    Args:
        wordListRDD (RDD of str): An RDD consisting of words.

    Returns:
        RDD of (str, int): An RDD consisting of (word, count) tuples.
    """
    return wordListRDD.map(lambda a: (a, 1)).reduceByKey(lambda a, b: a + b)

# 将读取到的文件转换为wordcount程序进行技术
top15WordsAndCounts = wordCount(shakeWordsRDD).takeOrdered(15, lambda a: -a[1])

# 这段程序说明流程 看不懂
print('\n'.join(map(lambda c: '{0}: {1}'.format(c[0], c[1]), top15WordsAndCounts)))
